#![allow(unused_mut)]
#![allow(unused_variables)]
#![allow(unused_imports)]
#![allow(unreachable_code)]
#![allow(dead_code)]
#![allow(unused_must_use)]

use std::collections::{HashMap, HashSet};
use std::convert::Infallible;
use std::env::var;
use std::net::{IpAddr, SocketAddr};
use std::time::Duration;
use std::{
    pin::Pin,
    task::{Context, Poll},
};

use async_std::fs;
use async_std::path;
use async_std::prelude::*;
use async_std::sync::{Arc, Mutex, RwLock};
use base64::{decode, encode};
use dotenv::dotenv;
use uuid::Uuid;

use online::online_server::{Online, OnlineServer};
use push::{AvroBulkCreate, JsonBulkCreate};
use signature::AccessToken;
use xdefault::DefaultReply;

use tonic::transport::{Channel, Server};
use tonic::{Request, Response, Status};

// use apache_avro::{from_avro_datum, Schema};
use avro_rs::{from_avro_datum, Schema};
use std::str;

use cdrs::authenticators::NoneAuthenticator;
use cdrs::cluster::session::new as new_session;
use cdrs::cluster::{ClusterTcpConfig, NodeTcpConfigBuilder};
use cdrs::load_balancing::RoundRobin;
use cdrs::query::*;
use cdrs::query_values;

use self::utils::*;
use rust_embed::RustEmbed;

#[derive(RustEmbed)]
#[folder = "meta-lib-static/avsc"]
struct AvscAsset;

pub mod xdefault {
    tonic::include_proto!("proto.xdefault");
}
pub mod signature {
    tonic::include_proto!("proto.signature");
}
pub mod push {
    tonic::include_proto!("proto.push");
}
pub mod online {
    tonic::include_proto!("proto.online");
}

#[derive(Default)]
pub struct MyOnline {}

#[tonic::async_trait]
impl Online for MyOnline {
    async fn create_avro_bulk(
        &self,
        create_request: Request<AvroBulkCreate>,
    ) -> Result<Response<DefaultReply>, Status> {
        let create = create_request.into_inner();
        if var("ONLINE_SERVER_AUTH_ENABLE") != Ok("false".to_string()) {
            auths::verify_token(create.header_authorization).map_err(|s| Status::unauthenticated(s))?;
        }
        let raw_bs = match &create.compression[..] {
            "zstd" => create.body,
            "gzip" => {
                codecs::gunzip_to_zstd(&create.body).map_err(|x| Status::failed_precondition(x))?
            }
            _ => Err(Status::invalid_argument(
                "unsupported compression type!".to_string(),
            ))?,
        };

        let mut bs = zstd::decode_all(raw_bs.as_slice())
            .map_err(|e| Status::unknown(format!("zstd decode error: {:?}", e)))?;

        let schema_bs = AvscAsset::get("dc_sdk_push_rust.avsc").expect("dc_sdk_dev.avsc not found!");
        let schema_s = str::from_utf8(&schema_bs).expect("");
        let schema = Schema::parse_str(&schema_s)
            .map_err(|e| Status::unknown(format!("schema parse error: {:?}", e)))?;
        let datum = from_avro_datum(&schema, &mut bs.as_slice(), None)
            .map_err(|e| Status::unknown(format!("avro read error: {:?}", e)))?;

        let ts = now_ts_sec();
        let uuid = Uuid::new_v4().to_string();
        let message = format!("{:?}", datum);
        let avro_b64 = base64::encode(bs);

        let online_prefix = var("ONLINE_SERVER_PREFIX").unwrap_or("data_online_dev".to_string());
        let cassandras =
            var("ONLINE_SERVER_CASSANDRA_HOSTS").unwrap_or("127.0.0.1:9042".to_string());

        let nodes = cassandras
            .split(",")
            .map(|host| NodeTcpConfigBuilder::new(host, NoneAuthenticator {}).build())
            .collect::<Vec<_>>();
        let cluster_config = ClusterTcpConfig(nodes);
        let no_compression =
            new_session(&cluster_config, RoundRobin::new()).expect("session should be created");

        let insert_stmt = format!(
            "INSERT INTO {}.event_log (ts, uuid, message, avro_b64) VALUES (?, ?, ?, ?)",
            online_prefix
        );
        no_compression
            .query_with_values(insert_stmt, query_values!(ts, uuid, message, avro_b64))
            .expect("insert table error");

        Ok(Response::new(DefaultReply {
            code: 0,
            message: "success".into(),
            details: "".into(),
            origin: "".into(),
        }))
    }
    async fn create_json_bulk(
        &self,
        create_request: Request<JsonBulkCreate>,
    ) -> Result<Response<DefaultReply>, Status> {
        let create = create_request.into_inner();
        if var("ONLINE_SERVER_AUTH_ENABLE") != Ok("false".to_string()) {
            auths::verify_token(create.header_authorization).map_err(|s| Status::unauthenticated(s))?;
        }
        let bs = create.body;
        let ts = now_ts_sec();

        let uuid = Uuid::new_v4().to_string();
        let message = str::from_utf8(&bs).expect("");

        let online_prefix = var("ONLINE_SERVER_PREFIX").unwrap_or("data_online_dev".to_string());
        let cassandras =
            var("ONLINE_SERVER_CASSANDRA_HOSTS").unwrap_or("127.0.0.1:9042".to_string());

        let nodes = cassandras
            .split(",")
            .map(|host| NodeTcpConfigBuilder::new(host, NoneAuthenticator {}).build())
            .collect::<Vec<_>>();
        let cluster_config = ClusterTcpConfig(nodes);
        let no_compression =
            new_session(&cluster_config, RoundRobin::new()).expect("session should be created");

        let insert_stmt = format!(
            "INSERT INTO {}.event_log (ts, uuid, message) VALUES (?, ?, ?)",
            online_prefix
        );
        no_compression
            .query_with_values(insert_stmt, query_values!(ts, uuid, message))
            .expect("insert table error");

        Ok(Response::new(DefaultReply {
            code: 0,
            message: "success".into(),
            details: "".into(),
            origin: "hold for ip".into(),
        }))
    }
}

mod auths {
    use super::*;
    use base64;

    use std::collections::HashMap;

    pub fn parse_access_token(s: &str) -> Result<AccessToken, String> {
        let access_token_m = s
            .split("&")
            .map(|x| utils::tuple2(x, "=").unwrap())
            .collect::<HashMap<&str, &str>>();

        let start_ts_s = access_token_m
            .get("start_ts")
            .ok_or("field start_ts not find!")?;
        let end_ts_s = access_token_m
            .get("end_ts")
            .ok_or("field end_ts not find!")?;

        let start_ts = u64::from_str_radix(start_ts_s, 10)
            .map_err(|_| String::from("start_ts is not a valid numeric!"))?;
        let end_ts = u64::from_str_radix(end_ts_s, 10)
            .map_err(|_| String::from("end_ts is not a valid numeric!"))?;

        let ip = access_token_m
            .get("ip")
            .ok_or("field ip not find!")?
            .to_string();
        let payload_enc = access_token_m
            .get("payload_enc")
            .ok_or("field payload_enc not find!")?
            .to_string();

        Ok(AccessToken {
            start_ts,
            end_ts,
            ip,
            payload_enc,
        })
    }
    pub fn verify_token(authorization: String) -> Result<(), String> {
        let pubkey_b64 = "3oJdJiBvUAbMCyX+Fw4+ttfvWhH7bKSrIpFDbvj+qz8=".to_string();
        let pubkey_bs = base64::decode(pubkey_b64).unwrap();
        let pubkey = codecs::pubkey(&pubkey_bs);

        let (token_type, signed_access_token) = utils::tuple2(&authorization, " ")
            .ok_or("token format error: [<token_type> <token_val>!")?;
        if token_type != "Bearer" {
            Err("Bearer token needed!")?;
        }

        let (sign, access_token_s) =
            utils::tuple2(&signed_access_token, "&").ok_or("missing sign and data!")?;

        let (sign_name, sign_b64) = utils::tuple2(&sign, "=").ok_or("key-value needed!")?;

        if sign_name != "sign" {
            Err("token sign is not first field!")?;
        }

        let sign_val = base64::decode(sign_b64).map_err(|_| "sign field error!".to_string())?;
        let check = pubkey.verify(&access_token_s.as_bytes(), &sign_val).is_ok();
        if !check {
            Err("token sign error!")?;
        }

        let access_token = parse_access_token(access_token_s)?;

        let now_ts = utils::now_ts_sec();

        if access_token.end_ts < now_ts {
            Err("token expire!")?;
        }
        Ok(())
    }
}

mod utils {
    use super::*;
    use chrono::prelude::*;
    use itertools::Itertools;
    use std::collections::HashMap;
    use std::time::{SystemTime, UNIX_EPOCH};

    pub fn now_str_trunc(seconds: u32) -> (String, String) {
        let china_dt = Utc::now().with_timezone(&FixedOffset::east(8 * 3600));
        let china_day_str = china_dt.date().format("%Y%m%d").to_string();

        let china_tm = china_dt.time();
        let china_tm_secs = china_tm.hour() * 3600 + china_tm.minute() * 60 + china_tm.second();
        let china_tm_trunc =
            NaiveTime::from_num_seconds_from_midnight(china_tm_secs / seconds * seconds, 0);

        let china_tm_str_trunc = china_tm_trunc.format("%H%M").to_string();
        (china_day_str, china_tm_str_trunc)
    }

    pub fn now_ts_sec() -> u64 {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_secs()
    }

    pub fn tuple2<'a>(s: &'a str, sep: &str) -> Option<(&'a str, &'a str)> {
        s.splitn(2, sep).next_tuple::<(_, _)>()
    }

    pub fn resolve_socket_addr(nodes: &str) -> Option<IpAddr> {
        let ips = get_if_addrs::get_if_addrs()
            .unwrap()
            .iter()
            .map(|interface| interface.ip())
            .collect::<HashSet<_>>();
        nodes
            .split(",")
            .filter_map(|x| x.parse::<IpAddr>().ok())
            .filter(|x| ips.contains(&x))
            .nth(0)
    }

    pub fn now_ts() -> u128 {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("Time went backwards")
            .as_millis()
    }

    pub fn generate_once() -> String {
        let my_uuid_s = Uuid::new_v4().to_string();
        let now_ts_s = now_ts().to_string();
        format!("{}_{}", now_ts_s, my_uuid_s)
    }
}

mod codecs {
    use super::*;
    use base64;
    use flate2::read::GzDecoder;
    use ring::signature::{UnparsedPublicKey, ED25519};
    use std::io::{Read, Write};

    use zstd::encode_all;

    pub fn gunzip_to_zstd(bs_zip: &[u8]) -> Result<Vec<u8>, String> {
        let mut gzip_decoder = GzDecoder::new(bs_zip);
        let mut bs = Vec::new();
        gzip_decoder
            .read_to_end(&mut bs)
            .map_err(|_| "gunzip error!".to_string())?;
        encode_all(&bs[..], 0).map_err(|_| "zstd encode error!".to_string())
    }
    pub fn pubkey(pubkey: &[u8]) -> UnparsedPublicKey<&[u8]> {
        UnparsedPublicKey::new(&ED25519, &pubkey)
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    dotenv::from_filename("meta-lib-static/env/.env").ok();

    let nodes = var("ONLINE_SERVER_NODES").unwrap_or("127.0.0.1".to_string());
    let port = var("ONLINE_SERVER_PORT").unwrap().parse::<u16>().unwrap();

    let o_ip = resolve_socket_addr(&nodes);
    if o_ip.is_none() {
        println!("ONLINE_SERVER_NODES not match! skip");
        return Ok(());
    }
    let addr = SocketAddr::new(o_ip.unwrap(), port);
    println!("start server at {}", addr);

    Server::builder()
        .add_service(OnlineServer::new(MyOnline::default()))
        .serve(addr)
        .await?;
    Ok(())
}
